/**
 * Copyright(c) 2015 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-8-1 下午12:44 Created by Daisy
 */
package com.jade.framework.mom.rocketmq;

import java.util.ArrayList;
import java.util.List;

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.jade.framework.mom.MomBrowser;
import com.jade.framework.mom.MomException;
import com.jade.framework.mom.MomMessage;

/**
 * 消息浏览器，直接使用消费者的接口。
 *
 * @author Daisy
 * @version 1.0 14-8-1 下午12:44
 */
public class RocketmqBrowser implements MomBrowser
{
    protected DefaultMQPushConsumer consumer;

    public void setConsumer(DefaultMQPushConsumer consumer)
    {
        this.consumer = consumer;
    }

    public MomMessage getMessage(String msgId)
            throws MomException
    {
        try {
            return RocketmqUtils.toMomMessage(consumer.viewMessage(msgId));
        }
        catch (Throwable t) {
            throw RocketmqUtils.handleException(t);
        }
    }

    public List<MomMessage> listMessage(String topic, String key, int maxNum)
            throws MomException
    {
        try {
            QueryResult result = consumer.queryMessage(topic, key, maxNum, 0, Integer.MAX_VALUE);
            List<MessageExt> msgs = result.getMessageList();
            List<MomMessage> momMsgs = new ArrayList<MomMessage>(msgs.size());
            for (MessageExt msg : msgs) {
                momMsgs.add(RocketmqUtils.toMomMessage(msg));
            }
            return momMsgs;
        }
        catch (Throwable t) {
            throw RocketmqUtils.handleException(t);
        }
    }

    public MomMessage getSingleMessage(String topic, String key)
            throws MomException
    {
        List<MomMessage> msgs = listMessage(topic, key, 1);
        if (msgs.isEmpty()) {
            return null;
        }
        else {
            return msgs.get(0);
        }
    }
}
